'''
@Date:2022/11/13  18:36
@Part：TTS
'''
import time
from concurrent.futures import ThreadPoolExecutor
import allure
from tools.common import *
from tools.get_headerjson import GetcJson
from conf.api_path import ApiPath
from tools.get_headerjson import GetHeaderjson
from mysqltest.mysql_db import sqlconnect
from tools.logger import log  # 导入封装的logger方法
import aiohttp
import asyncio
import pytest
import allure
import json
import sqlite3
import paho.mqtt.client as mqtt



ApiPath = ApiPath()
GetcJson = GetcJson()
GetHeaderjson = GetHeaderjson()

# MQTT代理服务器地址和端口
broker = "172.16.1.29"
port = 18832
username = "intranet"
password = "gst@2024_123!"
topic = "IOT/intranet/client/request/prod/#"
ids = []


"""
@allure.epic("GST-OpenApi-AutoTest")
@allure.feature("设备操作")
class audioStop:
  # 实例化对象
  @allure.story("停止播放")
  @allure.title("停止播放")
  @allure.description("停止当前播放")
  @pytest.mark.run(order=1)  # 用例执行顺序
  def audiostop(self):
    print("============audioStop—case1===========")
    with allure.step("step1:请求停止播放接口"):
      async def main():
        # 创建客户端实例
        client = mqtt.Client()
        # a = audioStop()
        # client.on_connect = on_connect

        # 连接到代理
        client.username_pw_set(username=username, password=password)
        client.connect(broker, port)

        sql1 = "select device_code,device_id from device where is_del = 0  and dept_id = 1"
        sql2 = "select id from mqtt_publish_record_2025  order by create_at desc limit 1"
        tupl = sqlconnect.sqljudje(sql1)
        pub_id = sqlconnect.sqljudje(sql2)
        pubishid = pub_id[0][0]
        len1 = len(tupl)

        for row in tupl:
          le = len(tupl)
          dev_code = str(row[0])  # 假设column_name是第一列
          device_id = str(row[1])  # 假设column_name是第一列
          topic = f"IOT/intranet/server/report/prod/{dev_code}"
          async with aiohttp.ClientSession() as session:
            # 创建异步任务，不等待结果
            task = asyncio.create_task(post_request(device_id))
            # 这里可以执行其他操作
            await asyncio.sleep(1)  # 例如，等待1秒后执行其他操作

            sql2 = "select id from mqtt_publish_record_2025  order by create_at desc limit 1"
            pub_id = sqlconnect.sqljudje(sql2)
            pubishid = pub_id[0][0]
            mess = {"cmd": "audioStop", "imei": dev_code, "publishId": pubishid, "result": "success", "resultId": 1,
                    "topic": topic}
            message = json.dumps(mess)
            client.publish(topic, message)
            await task
            # print("主程序继续执行...")
            log.info("设备端模拟回调的消息为:{}".format({message}))
            print("------------------------------------------")
        # 启动网络循环（异步）
        client.loop_start()

      #return main()
"""
#验证MQTT是否连接成功
def on_connect(client, userdata, flags,rc):
  if rc == 0:
    print("成功连接到MQTT代理！")
    client.subscribe(topic)
    print("打印订阅到的topic:", topic)
  else:
    print(f"连接失败，错误码：{rc}")



async def post_request(device_id):
  async with aiohttp.ClientSession() as session:

    GetcJson.set_audiostop_deviceId(device_id)
    
    
    async with session.post(url=ApiPath.audio_stop,
                            headers=GetHeaderjson.get_s_headers(),
                            json=GetcJson.get_audiostop()) as response:

      log.info("请求参数为:{},{}".format(ApiPath.audio_stop, GetcJson.get_audiostop()))
      log.info("响应结果为:{}".format(await response.text()))










async def main():
  # 创建客户端实例
  client = mqtt.Client()
  # 连接到代理
  client.username_pw_set(username=username, password=password)
  client.connect(broker,port)


  sql1 = "select device_code,device_id from device where is_del = 0  and dept_id = 1"
  sql2 = "select id from mqtt_publish_record_2025  order by create_at desc limit 1"
  tupl = sqlconnect.sqljudje(sql1)
  pub_id = sqlconnect.sqljudje(sql2)
  pubishid = pub_id[0][0]
  len1 = len(tupl)

  for row in tupl:
    le = len(tupl)
    dev_code = str(row[0])  # 假设column_name是第一列
    device_id = str(row[1])  # 假设column_name是第一列

    # print("打印deviceId:",device_id)
    ids.append(device_id)
    #print("输出code、id:",dev_code,device_id)
    #print(pubishid)
    #global topic
    topic = f"IOT/intranet/server/report/prod/{dev_code}"
    client.on_connect = on_connect
    #订阅
    #client.subscribe(topic)


    async with aiohttp.ClientSession() as session:
      # 创建异步任务，不等待结果
      task = asyncio.create_task(post_request(device_id))

      # 这里可以执行其他操作
      await asyncio.sleep(1)  # 例如，等待1秒后执行其他操作

      #获取回调信息
      #sql2 = "select id from mqtt_publish_record_2025  order by create_at desc limit 1"
      #pub_id = sqlconnect.sqljudje(sql2)
      #pubishid = pub_id[0][0]
      mess = {"cmd": "audioStop", "imei": dev_code, "publishId": pubishid, "result": "success", "resultId": 1,"topic": topic}
      message = json.dumps(mess)

      #模拟设备返回
      client.publish(topic, message)

      #等待异步任务执行结束
      await task
      #print("主程序继续执行...")
      log.info("设备端模拟回调的消息为:{}".format({message}))
      print("------------------------------------------")

  # 启动网络循环（异步）
  client.loop_start()


# Python 3.7及以上版本可以使用以下方式运行异步主函数
if __name__ == '__main__':
  asyncio.run(main())


